package com.exd.flink.stream.flow;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import com.exd.flink.common.HbaseTablesConstant;
import com.exd.flink.common.OutputVComputeTagBase;
import com.exd.flink.common.PropertiesTool;
import com.exd.flink.domain.AlgorithmStatusBean;
import com.exd.flink.domain.ResourcesBean;
import com.exd.flink.domain.VComputeBean;
import com.exd.flink.domain.VersionBean;
import com.exd.flink.sink.clickhouse.ClickHouseSink;
import com.exd.flink.sink.hbase.HbaseSink;
import com.exd.flink.stream.function.VcomputeAnalyticalFunction;

/**
 * 描述: 解析BdcmVcompute数据
 * 创建人: 赵兴炎
 * 日期: 2022年9月13日
 */
public class VcomputeStream {
	
	/**
	 * 解析BdcmVcompute数据
	 */
	public static void analyticalData(DataStream<String> bdcmStream,DataStream<String> vdcmStream) {
		// 解析BDCM数据
		SingleOutputStreamOperator<VComputeBean> bdcmStreamProcess = bdcmStream.process(new VcomputeAnalyticalFunction());
		// 获取vcompute数据
		DataStream<VComputeBean> bdcmVcomputeStream = bdcmStreamProcess.getSideOutput(OutputVComputeTagBase.BDCM_ANALYTICAL_VCOMPUTE_OK);
		// 获取版本数据
		DataStream<VersionBean> bdcmVersionStream = bdcmStreamProcess.getSideOutput(OutputVComputeTagBase.VERSION_DATA);
		// 获取资源监控数据
		DataStream<ResourcesBean> bdcmResourcesStream = bdcmStreamProcess.getSideOutput(OutputVComputeTagBase.RESOUCRES_DATA);
		// 获取算法状态数据
		DataStream<AlgorithmStatusBean> bdcmStatusStream = bdcmStreamProcess.getSideOutput(OutputVComputeTagBase.ALGORITHM_STATUS_DATA);
		
		// 解析VDCM数据
		SingleOutputStreamOperator<VComputeBean> vdcmStreamProcess = vdcmStream.process(new VcomputeAnalyticalFunction());
		// 获取vcompute数据
		DataStream<VComputeBean> vdcmVcomputeStream = vdcmStreamProcess.getSideOutput(OutputVComputeTagBase.VDCM_ANALYTICAL_VCOMPUTE_OK);
		// 获取版本数据
		DataStream<VersionBean> vdcmVersionStream = vdcmStreamProcess.getSideOutput(OutputVComputeTagBase.VERSION_DATA);
		// 获取资源监控数据
		DataStream<ResourcesBean> vdcmResourcesStream = vdcmStreamProcess.getSideOutput(OutputVComputeTagBase.RESOUCRES_DATA);
		// 获取算法状态数据
		DataStream<AlgorithmStatusBean>vdcmStatusStream = vdcmStreamProcess.getSideOutput(OutputVComputeTagBase.ALGORITHM_STATUS_DATA);
		
		//版本数据流合并
		DataStream<VersionBean> versionStream = bdcmVersionStream.union(vdcmVersionStream);
		//资源监控数据流合并
		DataStream<ResourcesBean> resourcesStream = bdcmResourcesStream.union(vdcmResourcesStream);
		//算法状态数据流合并
		DataStream<AlgorithmStatusBean> algorithmStatusStream = bdcmStatusStream.union(vdcmStatusStream);
		// 判断是否进行持久化操作
		if(PropertiesTool.getParameterTool().getBoolean("exd.stroage.flag")) {
			// BDCM vcompute 数据
			HbaseSink.vcomputeData(bdcmVcomputeStream, HbaseTablesConstant.BDCM_VCOMPUTE_PACKAGE);
			// VDCM vcompute 数据
			HbaseSink.vcomputeData(vdcmVcomputeStream, HbaseTablesConstant.VDCM_VCOMPUTE_PACKAGE);
			// 资源监控
			ClickHouseSink.inserResourcesData(resourcesStream);
			// 版本数据
//			MysqlSink.inserVersionData(versionStream);
			// 算法状态
//			MysqlSink.updateAlgorithmStatusData(algorithmStatusStream);
		}else {
			versionStream.print("version:");
			resourcesStream.print("resources:");
			algorithmStatusStream.print("algorithmStatus:");
			bdcmVcomputeStream.print("bcdmCompute:");
			vdcmVcomputeStream.print("vcdmCompute:");		
		}
	}

}
